package eg.ci.mvn.service.core.config.rabbitmq;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.util.ErrorHandler;

import java.net.URI;
import java.net.URISyntaxException;
import java.time.format.DateTimeParseException;
import java.util.concurrent.ThreadPoolExecutor;

/**
 *  定义一个 <code>test</code> 的 Rabbit 连接配置类 <br/>
 *  自行修改工厂参数 <br/>
 *  Producer(生产者) :生产消息的一方（邮件投递者） Spring中，采用 rabbitTemplate 来发送消息 <br/>
 *  Consumer(消费者) :消费消息的一方（邮件收件人） Spring中，采用 @RabbitListener 注解来接收消息 <br/>
 *
 *  Producer(生产者) -> Exchange(交换器) -> Queue(消息队列) -> Broker（消息中间件的服务节点） -> Consumer(消费者)
 *
 *  Exchange(交换器) 有 4 种类型，不同的类型对应着不同的路由策略：direct(默认)，fanout, topic, 和 headers
 */
@Slf4j
@Configuration
@ConditionalOnProperty(prefix = "rabbit.test", name = "uri", havingValue = "amqp://username:password@127.0.0.1:5672")
public class MqTestRabbitFactory {

    private static final String BEAN_NAME_PREFIX = "test";

    public static final String BEAN_NAME_RABBIT_ADMIN = BEAN_NAME_PREFIX + "RabbitAdmin";

    public static final String BEAN_NAME_RABBIT_TEMPLATE = BEAN_NAME_PREFIX + "RabbitTemplate";
    public static final String BEAN_NAME_RABBIT_TEMPLATE_CONFIRM_CALLBACK = BEAN_NAME_PREFIX + "RabbitTemplateConfirmCallback";
    public static final String BEAN_NAME_RABBIT_TEMPLATE_RETURNS_CALLBACK = BEAN_NAME_PREFIX + "RabbitTemplateReturnsCallback";

    public static final String BEAN_NAME_RABBIT_LISTENER_CONTAINER_FACTORY = BEAN_NAME_PREFIX + "MqListenerContainerFactory";
    public static final String BEAN_NAME_RABBIT_CONNECTION_FACTORY = BEAN_NAME_PREFIX + "MqConnectionFactory";
    public static final String BEAN_NAME_RABBIT_TASK_EXECUTOR = BEAN_NAME_PREFIX + "MqThreadPoolExecutor";

    public static final String BEAN_NAME_RABBIT_ERROR_HANDLER = BEAN_NAME_PREFIX + "MqErrorHandler";

    @Value("${rabbit.test.uri:amqp://username:password@127.0.0.1:5672}")
    private String rabbitUri;

    // RabbitMQ的使用入口
    @Bean(BEAN_NAME_RABBIT_TEMPLATE)
//    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    //必须是prototype类型
    public RabbitTemplate rabbitTemplate(
            @Qualifier(BEAN_NAME_RABBIT_CONNECTION_FACTORY) AbstractConnectionFactory connectionFactory,
            @Qualifier(BEAN_NAME_RABBIT_TEMPLATE_CONFIRM_CALLBACK) RabbitTemplate.ConfirmCallback confirmCallback,
            @Qualifier(BEAN_NAME_RABBIT_TEMPLATE_RETURNS_CALLBACK) RabbitTemplate.ReturnsCallback returnsCallback
            ) {
        // RabbitTemplate
        RabbitTemplate template = new RabbitTemplate();
        template.setConnectionFactory(connectionFactory);
        // Message Converter
        template.setMessageConverter(new Jackson2JsonMessageConverter());
        // Set the mandatory flag when sending messages;
        template.setMandatory(true);
        // Set confirmCallback
        template.setConfirmCallback(confirmCallback);
        // Set returnsCallback
        template.setReturnsCallback(returnsCallback);
        return template;
    }

    // RabbitMQ的使用-confirmCallback
    @Bean(BEAN_NAME_RABBIT_TEMPLATE_CONFIRM_CALLBACK)
    public RabbitTemplate.ConfirmCallback rabbitTemplateConfirmCallback() {
        return new ConfirmCallback();
    }

    // RabbitMQ的使用-confirmCallback
    @Bean(BEAN_NAME_RABBIT_TEMPLATE_RETURNS_CALLBACK)
    public RabbitTemplate.ReturnsCallback rabbitTemplateReturnsCallback() {
        return new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returned) {
                log.trace("{}", returned);
            }
        };
    }


    @Bean(BEAN_NAME_RABBIT_ADMIN)
    public RabbitAdmin rabbitAdmin(@Qualifier(BEAN_NAME_RABBIT_CONNECTION_FACTORY) ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }


    @Bean(BEAN_NAME_RABBIT_LISTENER_CONTAINER_FACTORY)
    public DirectRabbitListenerContainerFactory createMqListenerContainerFactory(
            @Qualifier(BEAN_NAME_RABBIT_CONNECTION_FACTORY) AbstractConnectionFactory connectionFactory,
            @Qualifier(BEAN_NAME_RABBIT_ERROR_HANDLER) ErrorHandler errorHandler
    ) {
        return createListenerContainerFactory(errorHandler, connectionFactory);
    }


    @Bean(BEAN_NAME_RABBIT_CONNECTION_FACTORY)
    public AbstractConnectionFactory connectionFactory(@Qualifier(BEAN_NAME_RABBIT_TASK_EXECUTOR) ThreadPoolExecutor executor) {
        return createConnectionFactory(rabbitUri, executor);
    }

    /**
     * 定义一个 异常处理用于注入 RabbitListener 的构造工厂的错误处理器
     *
     * @return 错误处理器使用自定义策略 {@link ConditionalRejectingErrorHandler}
     */
    @Bean(BEAN_NAME_RABBIT_ERROR_HANDLER)
    public ConditionalRejectingErrorHandler errorHandler() {
        return new ConditionalRejectingErrorHandler(new ConditionalRejectingErrorHandler.DefaultExceptionStrategy() {
            @Override
            protected boolean isUserCauseFatal(Throwable cause) {
                return cause instanceof NullPointerException
                        || cause instanceof IllegalStateException
                        || cause instanceof DateTimeParseException
//                    || cause instanceof BadSqlGrammarException
                        || cause instanceof IllegalArgumentException;
            }
        });
    }

    private AbstractConnectionFactory createConnectionFactory(String url, ThreadPoolExecutor executor) {
        URI uri;
        try {
            uri = new URI(url);
        } catch (URISyntaxException e) {
            throw new IllegalStateException("cloud mq uri error.", e);
        }
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(uri);
        // 10s 连接超时
        connectionFactory.setConnectionTimeout(10_000);
        // 10s 心跳
        connectionFactory.setRequestedHeartBeat(30);
//        StandardMetricsCollector metricsCollector = new StandardMetricsCollector(metricRegistry);
//        connectionFactory.getRabbitConnectionFactory().setMetricsCollector(metricsCollector);

        // # 新版本使用这个即可 confirm模式
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        // # 老版本的话用下面这个即可开启confirm确认模式
        connectionFactory.setPublisherConfirms(true);
        // # 开启确认模式
        connectionFactory.setPublisherReturns(true);

        connectionFactory.setExecutor(executor);

        connectionFactory.getRabbitConnectionFactory().getClientProperties().put("guo", "1");
        return connectionFactory;
    }

    /**
     * 往abstractConnectionFactory里面设置线程池
     * 这里需要注意 配置文件      rabbitmqTp对应的threadPoolName 要与  RabbitMqDtpAdapter的genTpName方法生成的名字对上
     */
    private DirectRabbitListenerContainerFactory createListenerContainerFactory(ErrorHandler errorHandler,
                                                                                AbstractConnectionFactory abstractConnectionFactory) {
        DirectRabbitListenerContainerFactory directFactory = new DirectRabbitListenerContainerFactory();
        // directFactory.setExposeListenerChannel(true);
        directFactory.setErrorHandler(errorHandler);
        directFactory.setConsumersPerQueue(1);
        // 每次处理数据数量，提高并发量
        directFactory.setPrefetchCount(1);
        directFactory.setIdleEventInterval(60_000L);
        directFactory.setConnectionFactory(abstractConnectionFactory);
        return directFactory;

    }

    //    static class RabbitTemplateWrapper extends org.springframework.amqp.rabbit.core.RabbitTemplate {
//        /**
//         * 默认必须带 CorrelationData
//         */
//        @Override
//        public void doSend(Channel channel, String exchangeArg, String routingKeyArg, Message message, boolean mandatory, CorrelationData correlationData) throws IOException {
//            super.doSend(channel, exchangeArg, routingKeyArg, message, mandatory, null == correlationData ? new CorrelationData(): correlationData);
//        }
//    }

    // @Slf4j
    public static class ConfirmCallback implements RabbitTemplate.ConfirmCallback {

        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            // 是否存在 correlationData
            boolean isExistsCorrelationData = null != correlationData;
            // 是否存在 correlationData.getReturned()
            boolean isExistsReturnMsg = null != correlationData && null != correlationData.getReturned();
            // 消息
            log.info("msgId: {},  ack: {}, cause: {}, replyCode: {}, replyText：{}, body: {}, returned: {}",
                    isExistsCorrelationData ? correlationData.getId() : "undefined",
                    ack,
                    cause,
                    isExistsReturnMsg ? correlationData.getReturned().getReplyCode() : "",
                    isExistsReturnMsg ? correlationData.getReturned().getReplyText() : "",
                    isExistsReturnMsg ? new String(correlationData.getReturned().getMessage().getBody()) : "",
                    isExistsReturnMsg ? correlationData.getReturned() : ""
            );

            if (isExistsCorrelationData && !ack) {
                log.info("rabbitmq未确认消息到达, 重发数据id:{}", correlationData.getId());
                // todo correlationData.getId 进行重发设置 重试次数
                return;
            }
        }
    }
}
